package com.sfx.shortLink.rmq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.lang.UUID;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.sfx.shortLink.common.constants.LinkConstant;
import com.sfx.shortLink.common.constants.RedisKeyConstant;
import com.sfx.shortLink.common.convention.exception.ServiceException;
import com.sfx.shortLink.dao.entity.*;
import com.sfx.shortLink.dao.mapper.*;
import com.sfx.shortLink.dto.ShortLinkStatsRecordDTO;
import com.sfx.shortLink.mq.idempotent.MessageIdempotentHandler;
import com.sfx.shortLink.rmq.config.RocketMQConsumerProperties;
import com.sfx.shortLink.rmq.topic.TopicConstants;
import lombok.RequiredArgsConstructor;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;

import static com.sfx.shortLink.rmq.topic.TopicConstants.SHORT_LINK_STATS_TOPIC_KEY;

@Component
@RequiredArgsConstructor
public class ShortLinkRmqStatsSaveConsumer implements InitializingBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(ShortLinkRmqStatsSaveConsumer.class);
    private final ILinkGotoMapper linkGotoMapper;
    private final ILinkAccessLogsMapper linkAccessLogsMapper;
    private final ILinkLocaleStatsMapper linkLocaleStatsMapper;
    private final ILinkAccessStatsMapper linkAccessStatsMapper;
    private final ILinkOsStatsMapper linkOsStatsMapper;
    private final ILinkBrowserMapper linkBrowserMapper;
    private final ILinkDeviceStatsMapper linkDeviceStatsMapper;
    private final ILinkNetworkStatsMapper linkNetworkStatsMapper;
    private final ILinkStatsTodayMapper linkStatsTodayMapper;
    private final ILinkMapper linkMapper;
    private final MessageIdempotentHandler messageIdempotentHandler;
    private final RedissonClient redissonClient;
    private final MQProducer mqProducer;

    @Value("${short-link.stats.locale.amap-key}")
    private String amapKey;

    private final RocketMQConsumerProperties rocketMQConsumerProperties;


    @Override
    public void afterPropertiesSet() throws Exception {
        DefaultMQPushConsumer mqPushConsumer = new DefaultMQPushConsumer();
        //老版本中会开启，新版本的mq不需要使用到
        mqPushConsumer.setVipChannelEnabled(false);
        mqPushConsumer.setNamesrvAddr(rocketMQConsumerProperties.getNameSrv());
        mqPushConsumer.setConsumerGroup(rocketMQConsumerProperties.getGroupName() + "_" + ShortLinkRmqStatsSaveConsumer.class.getSimpleName());
        //一次从broker中拉取10条消息到本地内存当中进行消费
        mqPushConsumer.setConsumeMessageBatchMaxSize(10);
        mqPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //监听im发送过来的业务消息topic
        mqPushConsumer.subscribe(SHORT_LINK_STATS_TOPIC_KEY, "*");
        mqPushConsumer.setMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                String msgId = msg.getMsgId();
                if (!messageIdempotentHandler.isConsume(msgId)) {
                    // 再检查一下看是否消费过
                    if (messageIdempotentHandler.isAccomplish(msgId)) {
                        // 消息已经消费完成
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    //消息消费出现了异常
                    throw new ServiceException("消息消费处理出现异常,无法被消费");
                }
                try {
                    Map<String, String> producerMap = JSON.parseObject(new String(msg.getBody()), Map.class);
                    String fullShortUrl = producerMap.get("fullShortUrl");
                    if (StrUtil.isNotBlank(fullShortUrl)) {
                        ShortLinkStatsRecordDTO statsRecordDTO = JSONUtil.parse(producerMap.get("statsRecordDTO")).toBean(ShortLinkStatsRecordDTO.class);
                        String gid = producerMap.get("gid");
                        //ShortLinkStatsRecordDTO statsRecordDTO = JSON.parseObject(producerMap.get("statsRecordDTO"), ShortLinkStatsRecordDTO.class);
                        // 统计短链接
                        shortLinkStats(fullShortUrl, gid, statsRecordDTO);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                } catch (Throwable throwable) {
                    messageIdempotentHandler.delMessage(msgId);
                }
                // 消息消费已完成
                messageIdempotentHandler.setAccomplish(msgId);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        mqPushConsumer.start();
        LOGGER.info("mq消费者启动成功,namesrv is {}", rocketMQConsumerProperties.getNameSrv());
    }

    public void shortLinkStats(String fullShortUrl, String gid, ShortLinkStatsRecordDTO statsRecordDTO) {
        fullShortUrl = Optional.ofNullable(fullShortUrl).orElse(statsRecordDTO.getFullShortUrl());
        // 获取读锁
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(RedisKeyConstant.LOCK_GID_UPDATE_KEY + fullShortUrl);
        RLock rLock = readWriteLock.readLock();
        if (!rLock.tryLock()) {
            statsRecordDTO.setKeys(UUID.fastUUID().toString());
            Map<String, Object> producerMap = new HashMap<>();
            producerMap.put("statsRecordDTO", statsRecordDTO);
            Message message = new Message();
            message.setBody(JSON.toJSONBytes(producerMap));
            message.setDelayTimeLevel(3);
            message.setTopic(TopicConstants.DELAY_QUEUE_STATS_KEY);
            try {
                SendResult sendResult = mqProducer.send(message);
                LOGGER.info("[ShortLinkRmqStatsSaveConsumer delayMessage] message is {}", sendResult);
            } catch (Exception e) {
                LOGGER.error("[ShortLinkRmqStatsSaveConsumer delayMessage] send message error", e);
            }
            return;
        }
        try {
            if (StrUtil.isBlank(gid)) {
                LambdaQueryWrapper<ShortLinkGotoDO> linkGotoQueryWrapper = Wrappers
                        .lambdaQuery(ShortLinkGotoDO.class)
                        .eq(ShortLinkGotoDO::getFullShortUrl, fullShortUrl);
                ShortLinkGotoDO shortLinkGotoDO = linkGotoMapper.selectOne(linkGotoQueryWrapper);
                gid = shortLinkGotoDO.getGid();
            }
            int hour = DateUtil.hour(new Date(), true);
            int weekDay = DateUtil.dayOfWeekEnum(new Date()).getValue();
            int uip = statsRecordDTO.isUipFlag() ? 1 : 0;
            int uv = statsRecordDTO.isUvFlag() ? 1 : 0;
            LinkAccessStatsDO linkAccessStatsDO = LinkAccessStatsDO.builder()
                    .date(new Date())
                    .pv(1)
                    .fullShortUrl(fullShortUrl)
                    .hour(hour)
                    .uip(uip)
                    .uv(uv)
                    .gid(gid)
                    .weekday(weekDay)
                    .build();
            linkAccessStatsMapper.shortLinkStatus(linkAccessStatsDO);

            Map<String, Object> paramMap = new HashMap<>();
            paramMap.put("ip", statsRecordDTO.getIp());
            paramMap.put("key", amapKey);
            String output = HttpUtil.get(LinkConstant.LINK_LOCAL_STATS_IP, paramMap);
            JSONObject jsonObject = JSON.parseObject(output);
            Integer infocode = jsonObject.getInteger("infocode");
            String city = "未知";
            String province = "未知";
            String adcode = "未知";
            String expectCity = jsonObject.getString("city");
            String expectAdcode = jsonObject.getString("adcode");
            String expectProvince = jsonObject.getString("province");
            if (infocode != null && infocode.equals(10000)) {
                LinkLocaleStatsDO linkLocaleStatsDO = LinkLocaleStatsDO.builder()
                        .country("中国")
                        .city(city = expectCity.equals("[]") ? city : expectCity)
                        .cnt(1)
                        .province(province = expectProvince.equals("[]") ? province : expectProvince)
                        .adcode(expectAdcode.equals("[]") ? adcode : expectAdcode)
                        .date(DateUtil.date())
                        .fullShortUrl(fullShortUrl)
                        .gid(gid)
                        .build();
                linkLocaleStatsMapper.LocaleStats(linkLocaleStatsDO);
            }

            LinkOsStatsDO linkOsStatsDO = LinkOsStatsDO.builder()
                    .os(statsRecordDTO.getOs())
                    .cnt(1)
                    .gid(gid)
                    .date(DateUtil.date())
                    .fullShortUrl(fullShortUrl)
                    .build();
            linkOsStatsMapper.osStats(linkOsStatsDO);

            LinkBrowserStatsDO browserStatsDO = LinkBrowserStatsDO.builder()
                    .browser(statsRecordDTO.getBrowser())
                    .cnt(1)
                    .date(DateUtil.date())
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .build();
            linkBrowserMapper.browserStats(browserStatsDO);

            LinkDeviceStatsDO linkDeviceStatsDO = LinkDeviceStatsDO.builder()
                    .cnt(1)
                    .date(DateUtil.date())
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .device(statsRecordDTO.getDevice())
                    .build();
            linkDeviceStatsMapper.shortLinkDeviceState(linkDeviceStatsDO);

            LinkNetworkStatsDO linkNetworkStatsDO = LinkNetworkStatsDO.builder()
                    .cnt(1)
                    .date(DateUtil.date())
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .network(statsRecordDTO.getNetwork())
                    .build();
            linkNetworkStatsMapper.shortLinkNetworkState(linkNetworkStatsDO);

            LinkAccessLogsDO linkAccessLogsDO = LinkAccessLogsDO.builder()
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .ip(statsRecordDTO.getIp())
                    .browser(statsRecordDTO.getBrowser())
                    .locale(StrUtil.join("-", "中国", province, city))
                    .os(statsRecordDTO.getOs())
                    .device(statsRecordDTO.getDevice())
                    .user(statsRecordDTO.getUser())
                    .network(statsRecordDTO.getNetwork())
                    .build();
            linkAccessLogsMapper.insert(linkAccessLogsDO);

            linkMapper.incrementStats(gid, fullShortUrl, 1, uv, uip, 1);

            LinkStatsTodayDO linkStatsTodayDO = LinkStatsTodayDO.builder()
                    .date(DateUtil.date())
                    .todayUv(uv)
                    .todayUip(uip)
                    .todayPv(1)
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .build();
            linkStatsTodayMapper.totayStats(linkStatsTodayDO);
        } catch (Exception e) {
            LOGGER.error("短链接统计异常");
            throw new ServiceException("短链接统计异常", e);
        } finally {
            rLock.unlock();
        }
    }
}